package org.wejob.core.log.util.mpscqueue;

import org.wejob.core.log.MessageRecord;

public class MpscArrayQueue<E extends MessageRecord> {
    private volatile long producerLimit;
    private volatile long producerIndex;
    private volatile long consumerIndex;
    private final static long P_LIMIT_OFFSET = UnsafeAccess.fieldOffset(MpscArrayQueue.class, "producerLimit");
    private final static long P_INDEX_OFFSET = UnsafeAccess.fieldOffset(MpscArrayQueue.class, "producerIndex");
    private final static long C_INDEX_OFFSET = UnsafeAccess.fieldOffset(MpscArrayQueue.class, "consumerIndex");

    protected final long mask;
    protected final E[] buffer;
    //需要提供内存检查
    public MpscArrayQueue(final int capacity)
    {
        this.producerLimit=capacity;
        int actualCapacity = roundToPowerOfTwo(capacity);
        mask = actualCapacity - 1;
        buffer = allocateRefArray(actualCapacity);
    }

    public boolean offer(final E e)
    {
        if (null == e)
        {
            throw new NullPointerException();
        }

        final long mask = this.mask;
        long producerLimit = lvProducerLimit();
        long pIndex;

        do
        {
            //获取producer index
            pIndex = lvProducerIndex();
            if (pIndex >= producerLimit)
            {
                //获取consumer index
                final long cIndex = lvConsumerIndex();
                //计算producer limit
                producerLimit = cIndex + mask + 1;
                if (pIndex >= producerLimit)
                {
                    return false;
                }
                else
                {
                    soProducerLimit(producerLimit);//这里对producerlimit的修改也是lazyset---底层调用了Unsafe类的putLongVolatile()方法
                }
            }
        }//如果cas抢占pIndex成功则使用该pIndex,否则重新获取pIndex并尝试抢占;
        while (!casProducerIndex(pIndex, pIndex + 1));
        e.setOffset(pIndex);
        final long offset = calcCircularRefElementOffset(pIndex, mask);
        UnsafeRefArrayAccess.soRefElement(buffer, offset, e);
        return true; // AWESOME :)
        //如果pIndex和offset分离,则cas获取锁之后入队并分配offset,然后释放锁;
        //单独维护offset的时候需要维护offset和消息之间的队列关系
    }
    public E poll()
    {
        final long cIndex = lpConsumerIndex();
        final long offset = calcCircularRefElementOffset(cIndex, mask);
        // Copy field to avoid re-reading after volatile load
        final E[] buffer = this.buffer;

        // If we can't see the next available element we can't poll
        E e = UnsafeRefArrayAccess.lvRefElement(buffer, offset);
        if (null == e)
        {
            /*
             * NOTE: Queue may not actually be empty in the case of a producer (P1) being interrupted after
             * winning the CAS on offer but before storing the element in the queue. Other producers may go on
             * to fill up the queue after this element.
             */
            if (cIndex != lvProducerIndex())
            {
                do
                {
                    e = UnsafeRefArrayAccess.lvRefElement(buffer, offset);
                }
                while (e == null);
            }
            else
            {
                return null;
            }
        }

        UnsafeRefArrayAccess.spRefElement(buffer, offset, null);
        soConsumerIndex(cIndex + 1);
        return e;
    }
    final long lvProducerLimit()
    {
        return producerLimit;
    }

    public final long lvProducerIndex()
    {
        return producerIndex;
    }

    public final long lvConsumerIndex()
    {
        return consumerIndex;
    }

    final void soProducerLimit(long newValue)
    {
        UnsafeAccess.UNSAFE.putOrderedLong(this, P_LIMIT_OFFSET, newValue);
    }

    final boolean casProducerIndex(long expect, long newValue)
    {
        return UnsafeAccess.UNSAFE.compareAndSwapLong(this, P_INDEX_OFFSET, expect, newValue);
    }
    public static long calcCircularRefElementOffset(long index, long mask)
    {
        return UnsafeRefArrayAccess.REF_ARRAY_BASE + ((index & mask) << UnsafeRefArrayAccess.REF_ELEMENT_SHIFT);
    }

    final void soConsumerIndex(long newValue)
    {
       UnsafeAccess.UNSAFE.putOrderedLong(this, C_INDEX_OFFSET, newValue);
    }

    final long lpConsumerIndex()
    {
        return org.jctools.util.UnsafeAccess.UNSAFE.getLong(this, C_INDEX_OFFSET);
    }

    private int roundToPowerOfTwo(int value) {
        if (value > 1073741824) {
            throw new IllegalArgumentException("There is no larger power of 2 int for value:" + value + " since it exceeds 2^31.");
        } else if (value < 0) {
            throw new IllegalArgumentException("Given value:" + value + ". Expecting value >= 0.");
        } else {
            int nextPow2 = 1 << 32 - Integer.numberOfLeadingZeros(value - 1);
            return nextPow2;
        }
    }

    private <E> E[] allocateRefArray(int capacity)
    {
        return (E[]) new MessageRecord[capacity];
    }
}
